package org.example.rabbitmqbatchack.test;

import com.alibaba.nacos.api.config.annotation.NacosValue;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;

@Component
public class CallCompleteConsumers {
    private Queue<Message> messageBuffer = new LinkedList<>();
    private long lastMessageTime = 0;  // 记录最后一条消息的时间
    @NacosValue(value = "${timeout:10000}", autoRefreshed = true)
    private  Long  timeout;
    @NacosValue(value = "${batchsize:5}", autoRefreshed = true)
    private  Integer  batchsize;

    // 消费者监听消息队列
    @RabbitListener(queues = "myconsumequeue")
    public void onMessage(List<Message> messages, Channel channel) throws IOException {
        // 每次收到消息时，更新最后接收的时间，并将消息存入缓存
        lastMessageTime = System.currentTimeMillis();
        for (Message message : messages) {
            messageBuffer.add(message);
        }
        // 批量确认如果消息数量超过 5 条，就立即处理
        if (messageBuffer.size() >= batchsize ) {
            processMessages(channel);
            for (Message message : messages) {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); // 确认每一条消息
            }
        }
    }
    // 定时检查队列，如果消息少于 5 条且 10秒内没有新消息，开始处理消息
    @Scheduled(fixedRate = 1000)  // 每秒检查一次
    public void checkAndProcessMessages() throws IOException {
        long currentTime = System.currentTimeMillis();
        // 如果队列中消息少于 5 条，且超过 10 秒没有新消息到来
        System.out.println(this.timeout);
        System.out.println(this.batchsize);
        if (messageBuffer.size() < 5 && currentTime - lastMessageTime > timeout) {
            processMessages(null);  // 这里我们不关心channel了，假设是批量处理
        }
    }
    // 处理消息
    private void processMessages(Channel channel) {
        if (messageBuffer.isEmpty()) {
            return;
        }
        System.out.println("开始处理消息...");
        // 处理缓存中的所有消息
        while (!messageBuffer.isEmpty()) {
            Message message = messageBuffer.poll();  // 从缓存队列中取出消息
            System.out.println(new String(message.getBody()));
        }
    }
}
